"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : Logical_Replication
Case Name   : wal_level=minimal时，创建和删除逻辑复制槽失败，合理报错
Create At   : 2023/9/18
Owner       : @xiaqi14
Description :
    1.创建目录，修改参数
    2.配置初始用户白名单
    3.wal_level=logical时，在主机上创建逻辑复制槽
    4.修改参数为wal_level=minimal
    5.wal_level=logical时，创建逻辑复制槽
    6.wal_level=minimal时，开启逻辑复制槽
    7.wal_level=minimal时，删除逻辑复制槽
    8.修改参数为wal_level=logical
    9.删除逻辑复制槽，删除目录
    10:资源清理
    11.恢复参数，恢复pg_hba.conf文件
Expect      :
    1.创建目录成功，修改参数成功
    2.配置成功
    3.创建成功
    4.修改成功
    5.创建失败，合理报错
    6.开启失败，合理报错
    7.删除失败，合理报错
    8.修改成功
    9.删除成功
    10:清理成功
    11.恢复成功
History     :
"""
import os
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.ComThread import ComThread

primary_sh = CommonSH('PrimaryDbUser')


@unittest.skipIf(1 != primary_sh.get_node_num(), '主备环境不执行')
class LogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.primary_node = Node('PrimaryDbUser')
        self.constant = Constant()
        self.common = Common()
        self.dir_replication = os.path.join(macro.DB_BACKUP_PATH,
                                            'dir_logical_replication_0056/')
        self.slot_name_1 = 'slot_logical_replication_0056_1'
        self.slot_name_2 = 'slot_logical_replication_0056_2'
        self.logfile_1 = os.path.join(self.dir_replication,
                                'logfile_logical_replication_0056_1.log')
        self.config_slot_log = 'enable_slot_log'
        self.slot_log_default = self.common.show_param(self.config_slot_log)
        self.config_wal_level = 'wal_level'
        self.wal_level_default = self.common.show_param(self.config_wal_level)
        self.archive_mode = 'archive_mode'
        self.archive_mode_default = self.common.show_param(self.archive_mode)
        self.hot_standby = 'hot_standby'
        self.hot_standby_default = self.common.show_param(self.hot_standby)
        self.max_wal_senders = 'max_wal_senders'
        self.max_wal_senders_default = \
            self.common.show_param(self.max_wal_senders)

    def test_logical_replication(self):
        text = "-----step1:创建目录，修改参数;" \
               "expect:创建目录成功，修改参数成功-----"
        self.logger.info(text)
        mkdir_cmd = f'''rm -rf {self.dir_replication} &&
                        mkdir {self.dir_replication} &&
                        ls -ld {self.dir_replication}'''
        primary_msg = self.primary_node.sh(mkdir_cmd).result()
        self.logger.info(primary_msg)
        self.assertIn(self.dir_replication, primary_msg, '执行失败' + text)

        if 'on' != self.slot_log_default:
            guc_result = primary_sh.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_slot_log}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.common.show_param(self.config_slot_log)
        self.assertEquals(show_msg, 'on', '执行失败' + text)

        if 'logical' != self.wal_level_default:
            guc_result = primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)
            restart_msg = primary_sh.restart_db_cluster()
            self.logger.info(restart_msg)
            self.assertTrue(restart_msg, '重启失败')
        show_msg = self.common.show_param(self.config_wal_level)
        self.assertEquals(show_msg, 'logical', '执行失败' + text)

        text = "-----step2:配置初始用户白名单;expect:配置成功-----"
        self.logger.info(text)
        param_1 = f'local replication {self.primary_node.ssh_user}  trust'
        param_2 = f'host  replication {self.primary_node.ssh_user} ' \
            f'127.0.0.1/32 trust'
        param_3 = f'host  replication {self.primary_node.ssh_user} ' \
            f'::1/128 trust'
        result_1 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_1)
        result_2 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_2)
        result_3 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text)

        text = "-----step3:wal_level=logical时，创建逻辑复制槽;expect:创建成功-----"
        self.logger.info(text)
        create_slot_msg = primary_sh.exec_pg_recvlogical(
            '--create', 'postgres', self.slot_name_1,
            self.primary_node.ssh_user, '-s 2')
        self.logger.info(create_slot_msg)
        text = "-----验证创建逻辑复制槽成功;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f"select * from pg_get_replication_slots() " \
            f"where slot_name like '%{self.slot_name_1}%';"
        sql_msg = primary_sh.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name_1, sql_msg, '执行失败' + text)

        text = "-----step4:修改参数为wal_level=minimal，expect:修改成功-----"
        self.logger.info(text)
        if 'off' != self.archive_mode_default:
            guc_result = primary_sh.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.archive_mode}=off')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.common.show_param(self.archive_mode)
        self.assertEquals(show_msg, 'off', '执行失败' + text)

        if 'off' != self.hot_standby_default:
            guc_result = primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.hot_standby}=off')
            self.assertTrue(guc_result, '执行失败' + text)

        if '0' != self.max_wal_senders_default:
            guc_result = primary_sh.execute_gsguc(
                 'set', self.constant.GSGUC_SUCCESS_MSG,
                 f'{self.max_wal_senders}=0')
            self.assertTrue(guc_result, '执行失败' + text)
        guc_result = primary_sh.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'{self.config_wal_level}=minimal')
        self.assertTrue(guc_result, '执行失败' + text)
        restart_msg = primary_sh.restart_db_cluster()
        self.logger.info(restart_msg)
        self.assertTrue(restart_msg, '重启失败')
        show_msg = self.common.show_param(self.config_wal_level)
        self.assertEquals(show_msg, 'minimal', '执行失败' + text)
        show_msg = self.common.show_param(self.hot_standby)
        self.assertEquals(show_msg, 'off', '执行失败' + text)
        show_msg = self.common.show_param(self.max_wal_senders)
        self.assertEquals(show_msg, '0', '执行失败' + text)

        text = "-----step5:wal_level=minimal时，创建逻辑复制槽;" \
               "expect:创建失败，合理报错-----"
        self.logger.info(text)
        primary_create_msg = primary_sh.exec_pg_recvlogical(
            '--create', 'postgres', self.slot_name_2,
            self.primary_node.ssh_user, '-s 2')
        self.logger.info(primary_create_msg)
        self.assertIn("FATAL:  number of requested standby connections "
                      "exceeds max_wal_senders (currently 0)",
                      primary_create_msg, '执行失败' + text)

        text = "-----step6:wal_level=minimal时，开启逻辑复制槽;" \
               "expect:开启失败，合理报错-----"
        self.logger.info(text)
        thread_1 = ComThread(primary_sh.exec_pg_recvlogical,
                             args=('--start', 'postgres', self.slot_name_1,
                                   self.primary_node.ssh_user,
                                   f'-f {self.logfile_1} '
                                   f'-o skip-empty-xacts=true -F 1 -s 2',))
        thread_1.setDaemon(True)
        thread_1.start()
        thread_1.join(10)
        self.logger.info(thread_1.get_result())
        self.assertIn("FATAL:  number of requested standby connections "
                      "exceeds max_wal_senders (currently 0)",
                      thread_1.get_result(), '执行失败' + text)

        text = "-----step7:wal_level=minimal时，删除逻辑复制槽;" \
               "expect:删除失败，合理报错-----"
        self.logger.info(text)
        primary_drop_msg = primary_sh.exec_pg_recvlogical(
            '--drop', 'postgres', self.slot_name_1,
            self.primary_node.ssh_user, '-s 2')
        self.logger.info(primary_drop_msg)
        self.assertIn("FATAL:  number of requested standby connections "
                      "exceeds max_wal_senders (currently 0)",
                      primary_drop_msg, '执行失败' + text)

        text = "-----step8:修改参数为wal_level=logical，expect:修改成功-----"
        self.logger.info(text)
        archive_mode_current = self.common.show_param(self.archive_mode)
        if archive_mode_current != self.archive_mode_default:
            guc_result = primary_sh.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.archive_mode}={self.archive_mode_default}')
            self.assertTrue(guc_result, '执行失败' + text)
        show_archive_mode = self.common.show_param(self.archive_mode)
        self.assertEquals(show_archive_mode, self.archive_mode_default,
                          '执行失败' + text)

        hot_standby_current = self.common.show_param(self.hot_standby)
        if hot_standby_current != self.hot_standby_default:
            guc_result = primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.hot_standby}={self.hot_standby_default}')
            self.assertTrue(guc_result, '执行失败' + text)

        max_wal_senders_current = self.common.show_param(self.max_wal_senders)
        if max_wal_senders_current != self.max_wal_senders_default:
            guc_result = primary_sh.execute_gsguc(
                 'set', self.constant.GSGUC_SUCCESS_MSG,
                 f'{self.max_wal_senders}={self.max_wal_senders_default}')
            self.assertTrue(guc_result, '执行失败' + text)

        guc_result = primary_sh.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'{self.config_wal_level}=logical')
        self.assertTrue(guc_result, '执行失败' + text)
        restart_msg = primary_sh.restart_db_cluster()
        self.logger.info(restart_msg)
        self.assertTrue(restart_msg, '重启失败')
        show_wal_level = self.common.show_param(self.config_wal_level)
        self.assertEquals(show_wal_level, 'logical', '执行失败' + text)
        show_hot_standby = self.common.show_param(self.hot_standby)
        self.assertEquals(show_hot_standby,
                          self.hot_standby_default, '执行失败' + text)
        show_max_wal_senders = self.common.show_param(self.max_wal_senders)
        self.assertEquals(show_max_wal_senders,
                          self.max_wal_senders_default, '执行失败' + text)

    def tearDown(self):
        text_1 = "-----step9:删除逻辑复制槽;expect:删除成功-----"
        self.logger.info(text_1)
        drop_slot_msg = primary_sh.exec_pg_recvlogical(
            '--drop', 'postgres', self.slot_name_1,
            self.primary_node.ssh_user, '-s 2')
        self.logger.info(drop_slot_msg)
        sql_cmd = f"select * from pg_get_replication_slots() " \
            f"where slot_name like '%slot_logical_replication_0056%';"
        sql_msg = primary_sh.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)

        text_2 = "-----step10:资源清理;expect:清理成功-----"
        self.logger.info(text_2)
        del_cmd = f'rm -rf {self.dir_replication};'
        self.logger.info(del_cmd)
        primary_del_res = self.primary_node.sh(del_cmd).result()
        self.logger.info(primary_del_res)
        file_cmd = f'''if [ -d {self.dir_replication} ];
            then echo "exists"; else echo "not exists"; fi'''
        primary_file_msg = self.primary_node.sh(file_cmd).result()

        text_3 = "-----step11:恢复参数，恢复pg_hba.conf文件，expect:恢复成功-----"
        self.logger.info(text_3)
        slot_log_current = self.common.show_param(self.config_slot_log)
        if slot_log_current != self.slot_log_default:
            primary_sh.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_slot_log}={self.slot_log_default}')
        show_slot_log = self.common.show_param(self.config_slot_log)

        param_1 = f'local replication {self.primary_node.ssh_user} '
        param_2 = f'host replication {self.primary_node.ssh_user} 127.0.0.1/32 '
        param_3 = f'host  replication {self.primary_node.ssh_user} ::1/128 '
        result_1 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_1)
        result_2 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_2)
        result_3 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_3)

        wal_level_current = self.common.show_param(self.config_wal_level)
        if wal_level_current != self.wal_level_default:
            primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}={self.wal_level_default}')
            restart_msg = primary_sh.restart_db_cluster()
            self.logger.info(restart_msg)
        show_wal_level = self.common.show_param(self.config_wal_level)
        self.assertIn("(0 rows)", sql_msg, '执行失败:' + text_1)
        self.assertEqual("not exists", primary_file_msg, '执行失败' + text_2)
        self.assertEquals(show_slot_log, self.slot_log_default,
                          '执行失败' + text_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text_3)
        self.assertEquals(show_wal_level, self.wal_level_default,
                          '执行失败' + text_3)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")
